package com.tzu.seq.bean;

import com.alipay.sofa.jraft.rhea.options.PlacementDriverOptions;
import com.alipay.sofa.jraft.rhea.options.RheaKVStoreOptions;
import com.alipay.sofa.jraft.rhea.options.StoreEngineOptions;
import com.alipay.sofa.jraft.rhea.options.configured.MemoryDBOptionsConfigured;
import com.alipay.sofa.jraft.rhea.options.configured.PlacementDriverOptionsConfigured;
import com.alipay.sofa.jraft.rhea.options.configured.RheaKVStoreOptionsConfigured;
import com.alipay.sofa.jraft.rhea.options.configured.StoreEngineOptionsConfigured;
import com.alipay.sofa.jraft.rhea.storage.StorageType;
import com.alipay.sofa.jraft.util.Endpoint;
import com.alipay.sofa.rpc.config.ConsumerConfig;
import com.alipay.sofa.rpc.listener.ChannelListener;
import com.alipay.sofa.rpc.transport.AbstractChannel;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.vertx.core.Vertx;
import io.vertx.core.datagram.DatagramSocket;
import io.vertx.core.datagram.DatagramSocketOptions;
import lombok.Getter;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.ToString;
import lombok.extern.log4j.Log4j2;
import thirdpart.codec.IBodyCodec;
import thirdpart.fetchsurv.IFetchService;

import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import java.util.Properties;
import java.util.Timer;

/**
 * @Author: Tzu
 * @Date: 2020/11/26 16:00
 * @Version: 1.0
 * @Description: 排队机启动入口/配置
 */
@Log4j2
@ToString
@RequiredArgsConstructor
public class SeqConfig {

    private String datapath;

    private String serveurlIp;

    private String serveurlPort;

    private String serverlist;

    @NonNull
    private InputStream in;

    //////////////////////排队机启动接口////////////////////////////////
    @Getter
    private Node node;

    public void startup() throws Exception{
        //1、读取配置文件
        initConfig();
        //2、初始化(启动) kv store
        startSeqDbCluster();
        //3、启动下游广播
        startMultiCast();
        //4、上游：初始化排队机-网关连接 定时从网关抓取订单数据
        startupFetch();
    }

    ///////////////////////////初始化/读取配置文件信息/////////////////////////////////////////////////
    /**
     * 初始化配置文件
     */
    private void initConfig() throws IOException {
        //加载配置文件
        Properties properties = new Properties();
//        properties.load(Object.class.getResourceAsStream("/"+fileName));
        properties.load(in);

        //加载配置属性的值到类成员上
        datapath = properties.getProperty("datapath");
        serveurlIp = properties.getProperty("serveurlIp");
        serveurlPort = properties.getProperty("serveurlPort");
        serverlist = properties.getProperty("serverlist");
        fetchurls = properties.getProperty("fetchurls");
        multicastIp = properties.getProperty("multicastip");
        multicastPort = Integer.parseInt(properties.getProperty("multicastport"));

        log.info("read config:{}",this);
    }

    ////////////////////////////////启动kvStore//////////////////////////////////////////////

    /**
     * 启动kvStore
     */
    private void startSeqDbCluster() {
        //.withFake(true)：多个集群时，这个参数为false，我们只有一个集群
        final PlacementDriverOptions pdOpts = PlacementDriverOptionsConfigured.newConfigured()
                .withFake(true)
                .config();

        final StoreEngineOptions storeOpts = StoreEngineOptionsConfigured.newConfigured()
                .withStorageType(StorageType.Memory)//使用内存做存储
                .withMemoryDBOptions(MemoryDBOptionsConfigured.newConfigured().config())
                .withRaftDataPath(datapath)//排队机集群日志信息文件输出地址
                .withServerAddress(new Endpoint(serveurlIp,Integer.parseInt(serveurlPort)))
                .config();

        final RheaKVStoreOptions opts = RheaKVStoreOptionsConfigured.newConfigured()
                .withInitialServerList(serverlist)//加载集群内机器地址
                .withStoreEngineOptions(storeOpts)
                .withPlacementDriverOptions(pdOpts)
                .config();

        node = new Node(opts);
        node.start();
        //把该节点的的关闭状态跟JDK的关闭绑定在一起
        Runtime.getRuntime().addShutdownHook(new Thread(node::stop));

        log.info("start seq node success on port : {}",serveurlPort);
    }


    //////////////////////广播：排队机==>撮合核心////////////////////////////////
    @Getter
    private String multicastIp;

    @Getter
    private int multicastPort;

    @Getter
    private DatagramSocket multicastSender;//发送udp报文工具类

    private void startMultiCast() {
        multicastSender = Vertx.vertx().createDatagramSocket(new DatagramSocketOptions());
    }


    //////////////////////从网关抓取订单信息逻辑////////////////////////////////

    private String fetchurls;

    //存储排队机和网关的连接记录
    @ToString.Exclude
    @Getter
    private Map<String, IFetchService> fetchServiceMap = Maps.newConcurrentMap();

    @NonNull
    @ToString.Exclude
    @Getter
    private IBodyCodec codec;

    //从哪些网关抓取
    //通信方式：rpc
    private void startupFetch() {
        //1、建立到所有网关的连接
        String[] urls = fetchurls.split(";");
        for (String url : urls) {
            //设置连接参数 连接网关
            ConsumerConfig<IFetchService> consumerConfig = new ConsumerConfig<IFetchService>()
                    .setInterfaceId(IFetchService.class.getName())//通信接口
                    .setProtocol("bolt")//RPC通信协议
                    .setTimeout(5000)//超时时间
                    .setDirectUrl(url);//直连地址
            //排队机只有在断开重连后才会进入setOnConnect()方法
            consumerConfig.setOnConnect(Lists.newArrayList(new FetchChannelListener(consumerConfig)));
            //排队机连接上网关以后，记录连接信息
            fetchServiceMap.put(url,consumerConfig.refer());
        }

        //2、创建定时从网关中抓取数据的任务
        new Timer().schedule(new FetchTask(this),5000,1000);
    }

    //监听器：监听排队机是否连接上网关并执行对应操作
    @RequiredArgsConstructor
    private class FetchChannelListener implements ChannelListener{

        @NonNull
        private ConsumerConfig<IFetchService> config;

        //连接上网关执行该方法（第一次连接上不会执行该方法，断开重连后才会执行该方法）
        @Override
        public void onConnected(AbstractChannel channel) {
            //排队机连接上的远程网关的地址
            String remoteAddr = channel.remoteAddress().toString();
            log.info("connect to gateway : {}",remoteAddr);
            //把排队机和网关的连接存在fetchServiceMap
            fetchServiceMap.put(remoteAddr,config.refer());
        }

        //与网关断开连接执行该方法
        @Override
        public void onDisconnected(AbstractChannel channel) {
            String remoteAddr = channel.remoteAddress().toString();
            log.info("disconnect from gateway : {}",remoteAddr);
            //与网关断开连接,移除fetchServiceMap里的连接信息
            fetchServiceMap.remove(remoteAddr);
        }
    }



    //////////////////////////////构造器//////////////////////////////////////////////////////////

    public SeqConfig() {
    }

    public SeqConfig(@NonNull InputStream in) {
        this.in = in;
    }
}
